// PassengerPatternMapper.java
package com.taxi.analysis.passengerTrip;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 乘客行程模式分析Mapper
 * 功能：解析乘客行程数据，生成用于二次排序的复合键
 * 技术：二次排序
 */
public class PassengerPatternMapper extends Mapper<LongWritable, Text, PassengerTripKey, Text> {

    private PassengerTripKey outputKey = new PassengerTripKey();
    private Text outputValue = new Text();
    private SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        // 跳过CSV标题行
        if (key.get() == 0) {
            return;
        }

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) {
                return;
            }

            // 解析CSV数据行
            String[] fields = parseCSVLine(line);

            // CSV字段索引（根据passenger_pattern_cleaning.py的输出格式）:
            // 0:passenger_id, 1:trip_sequence, 2:tpep_pickup_datetime, 3:tpep_dropoff_datetime,
            // 4:pickup_zone, 5:dropoff_zone, 6:trip_distance, 7:total_amount, 8:trip_duration,
            // 9:time_gap_minutes, 10:time_period, 11:trip_type, 12:fare_level, 13:passenger_type,
            // 14:total_trips, 15:avg_spending, 16:is_continuous_trip

            if (fields.length >= 17) {
                // 提取关键字段
                String passengerId = fields[0].trim();
                int tripSequence = parseInt(fields[1]);
                String pickupDatetime = fields[2].trim();
                String dropoffDatetime = fields[3].trim();
                String pickupZone = fields[4].trim();
                String dropoffZone = fields[5].trim();
                double tripDistance = parseDouble(fields[6]);
                double totalAmount = parseDouble(fields[7]);
                double tripDuration = parseDouble(fields[8]);
                double timeGapMinutes = parseDouble(fields[9]);
                String timePeriod = fields[10].trim();
                String tripType = fields[11].trim();
                String fareLevel = fields[12].trim();
                String passengerType = fields[13].trim();
                int totalTrips = parseInt(fields[14]);
                double avgSpending = parseDouble(fields[15]);
                boolean isContinuousTrip = parseBoolean(fields[16]);

                // 验证数据有效性
                if (isValidTripData(passengerId, pickupDatetime, pickupZone, dropoffZone)) {

                    // 解析时间戳
                    long pickupTimestamp = parseTimestamp(pickupDatetime);

                    // 创建复合键用于二次排序
                    outputKey.setPassengerId(passengerId);
                    outputKey.setPickupTimestamp(pickupTimestamp);
                    outputKey.setTripSequence(tripSequence);

                    // 创建包含所有行程信息的值
                    String tripInfo = createTripInfo(tripSequence, pickupDatetime, dropoffDatetime,
                            pickupZone, dropoffZone, tripDistance, totalAmount,
                            tripDuration, timeGapMinutes, timePeriod, tripType,
                            fareLevel, passengerType, totalTrips, avgSpending,
                            isContinuousTrip);

                    outputValue.set(tripInfo);
                    context.write(outputKey, outputValue);
                }
            }

        } catch (Exception e) {
            // 记录错误但继续处理其他行
            System.err.println("处理行时出错: " + value.toString());
            System.err.println("错误信息: " + e.getMessage());
        }
    }

    /**
     * 验证行程数据的有效性
     */
    private boolean isValidTripData(String passengerId, String pickupDatetime,
                                    String pickupZone, String dropoffZone) {
        return !passengerId.isEmpty() &&
                !pickupDatetime.isEmpty() &&
                !pickupZone.isEmpty() &&
                !dropoffZone.isEmpty() &&
                !passengerId.equals("null") &&
                !pickupZone.equals("null") &&
                !dropoffZone.equals("null");
    }

    /**
     * 解析时间戳
     */
    private long parseTimestamp(String datetime) {
        try {
            // 处理不同的时间格式
            if (datetime.contains("T")) {
                // ISO格式: 2023-01-01T12:00:00
                datetime = datetime.replace("T", " ");
                if (datetime.contains("+")) {
                    datetime = datetime.substring(0, datetime.indexOf("+"));
                }
            }

            // 移除可能的毫秒部分
            if (datetime.contains(".")) {
                datetime = datetime.substring(0, datetime.indexOf("."));
            }

            Date date = dateFormat.parse(datetime);
            return date.getTime();
        } catch (ParseException e) {
            System.err.println("时间解析错误: " + datetime);
            return 0L;
        }
    }

    /**
     * 创建行程信息字符串
     */
    private String createTripInfo(int tripSequence, String pickupDatetime, String dropoffDatetime,
                                  String pickupZone, String dropoffZone, double tripDistance,
                                  double totalAmount, double tripDuration, double timeGapMinutes,
                                  String timePeriod, String tripType, String fareLevel,
                                  String passengerType, int totalTrips, double avgSpending,
                                  boolean isContinuousTrip) {

        return String.format("%d|%s|%s|%s|%s|%.2f|%.2f|%.2f|%.2f|%s|%s|%s|%s|%d|%.2f|%s",
                tripSequence, pickupDatetime, dropoffDatetime, pickupZone, dropoffZone,
                tripDistance, totalAmount, tripDuration, timeGapMinutes, timePeriod,
                tripType, fareLevel, passengerType, totalTrips, avgSpending,
                isContinuousTrip ? "是" : "否");
    }

    /**
     * 解析CSV行，处理可能的引号和逗号
     */
    private String[] parseCSVLine(String line) {
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    /**
     * 处理包含引号的CSV行
     */
    private String[] parseQuotedCSV(String line) {
        java.util.List<String> fields = new java.util.ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }

    /**
     * 安全的double解析
     */
    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    /**
     * 安全的int解析
     */
    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }

    /**
     * 安全的boolean解析
     */
    private boolean parseBoolean(String str) {
        try {
            return Boolean.parseBoolean(str.trim()) ||
                    "True".equalsIgnoreCase(str.trim()) ||
                    "是".equals(str.trim());
        } catch (Exception e) {
            return false;
        }
    }
}
